package com.atom.zx;

import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import com.atom.zx.concurrency.TestScheduler;
import com.atom.zx.subscriptions.Subscriptions;
import com.atom.zx.util.Logs;
import com.atom.zx.util.functions.Action0;
import com.atom.zx.util.functions.Func1;
import com.atom.zx.util.functions.Func2;

/**
 * 调度程序
 */
public abstract class Scheduler {

    public abstract <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action);

    public abstract <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit);

    public <T> Subscription schedulePeriodically(T state, final Func2<Scheduler, T, Subscription> action, long initialDelay, long period, TimeUnit unit) {
        // 根据时间单位获取定时的时间
        final long periodInNanos = unit.toNanos(period);
        // 完成标志位
        final AtomicBoolean complete = new AtomicBoolean();
        // 创建一个方法 递归的方法
        final Func2<Scheduler, T, Subscription> recursiveAction = new Func2<Scheduler, T, Subscription>() {
            @Override
            public Subscription call(Scheduler scheduler, T state0) {
                // 判断是否完成
                if (!complete.get()) {
                    // 没有完成 获取开始时间戳
                    long startedAt = now();
                    // 进行执行 并获取订阅器
                    final Subscription sub1 = action.call(scheduler, state0);
                    // 获取执行花费时间
                    long timeTakenByActionInNanos = TimeUnit.MILLISECONDS.toNanos(now() - startedAt);
                    // 根据 状态 活动 定时时间和执行时间的差值 时间单位 创建一个订阅器
                    final Subscription sub2 = schedule(state0, this, periodInNanos - timeTakenByActionInNanos, TimeUnit.NANOSECONDS);
                    return Subscriptions.create(new Action0() {
                        @Override
                        public void call() {
                            sub1.unsubscribe(); // 退订
                            sub2.unsubscribe(); // 退订
                        }
                    });
                }
                return Subscriptions.empty();
            }
        };
        // 在test Scheduler中进行包装为timeAction后 入队操作 , 队列中还需要进行排队
        final Subscription sub = schedule(state, recursiveAction, initialDelay, unit);
        // 返回一个持有订阅开关的订阅器 , 可以提前进行退订
        return Subscriptions.create(new Action0() {
            @Override
            public void call() {
                complete.set(true);
                sub.unsubscribe();
            }
        });
    }

    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, Date dueTime) {
        long scheduledTime = dueTime.getTime();
        long timeInFuture = scheduledTime - now();
        if (timeInFuture <= 0) {
            return schedule(state, action);
        } else {
            return schedule(state, action, timeInFuture, TimeUnit.MILLISECONDS);
        }
    }

    public Subscription schedule(final Action0 action) {
        return schedule(null, new Func2<Scheduler, Void, Subscription>() {

            @Override
            public Subscription call(@SuppressWarnings("unused") Scheduler scheduler, @SuppressWarnings("unused") Void state) {
                action.call();
                return Subscriptions.empty();
            }
        });
    }

    public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
        return schedule(null, new Func2<Scheduler, Void, Subscription>() {

            @Override
            public Subscription call(@SuppressWarnings("unused") Scheduler scheduler, @SuppressWarnings("unused") Void state) {
                action.call();
                return Subscriptions.empty();
            }
        }, delayTime, unit);
    }

    public Subscription schedulePeriodically(final Action0 action, long initialDelay, long period, TimeUnit unit) {
        return schedulePeriodically(null, new Func2<Scheduler, Void, Subscription>() {
            @Override
            public Subscription call(@SuppressWarnings("unused") Scheduler scheduler, @SuppressWarnings("unused") Void state) {
                action.call();
                return Subscriptions.empty();
            }
        }, initialDelay, period, unit);
    }

    public long now() {
        return System.currentTimeMillis();
    }

    public static class UnitTest {

        public void testPeriodicScheduling() {
            // 创建一个func 无返回的执行单元
            final Func1<Long, Void> calledOp = aLong -> {
                Logs.e("Run Action Func1<Long, Void> "+aLong);
                return null;
            };
            // 创建一个任务调度器
            final TestScheduler scheduler = new TestScheduler();
            // 根据调度去创建一个具有定期的 订阅器
            Subscription subscription = scheduler.schedulePeriodically(new Action0() {
                @Override
                public void call() {
                    Logs.e(scheduler.now());
                    calledOp.call(scheduler.now());
                }
            }, 1, 2, TimeUnit.SECONDS);
            // 延时一秒  循环2秒
            // 1000000000  2000000000
            Logs.e("\n1 -> 999000000 纳秒"); // 秒 -- > 毫秒 --> 微秒 --> 纳秒
            scheduler.advanceTimeBy(999L, TimeUnit.MILLISECONDS);
            Logs.e("\n2 -> 1000000000");
            scheduler.advanceTimeBy(1L, TimeUnit.MILLISECONDS);
            Logs.e("\n3 -> 2999000000");
            scheduler.advanceTimeBy(1999L, TimeUnit.MILLISECONDS);
            Logs.e("\n4 -> ");
            scheduler.advanceTimeBy(1L, TimeUnit.MILLISECONDS);
            Logs.e("\n5 ->");
            scheduler.advanceTimeBy(5L, TimeUnit.SECONDS);
            Logs.e("\n6 -> ");
            subscription.unsubscribe();
            scheduler.advanceTimeBy(11L, TimeUnit.SECONDS);
        }
    }

    public static void main(String[] bb) {
        Scheduler.UnitTest test = new Scheduler.UnitTest();
        test.testPeriodicScheduling();
    }
}
